package member

import (
	"context"
	"fmt"
	"phant-operator/pkg/apis/label"
	"phant-operator/pkg/controller"
	"sync"
	"time"

	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/klog/v2"
)

// BackupTracker implements the logic for tracking endpoints created by partoni with DCS type kubernetes
type PatroniTracker interface {
	Length() int
	StartTrackPatroniEndPoints() error
	StartTrackPostgresCluster(ns, pgName string)
}

// the main processes of log backup track:
// a. tracker endpoints for PostgresCluster which are created by patroni
// b. tracker postgrescluster delete list, if not found, try to delete endpoints
// c. after tracker endpoints deleted, update postgrescluster delete list

type patroniTracker struct {
	deps         *controller.Dependencies
	operateLock  sync.Mutex
	waitDuration time.Duration

	// if PostgresCluster CR deleted, add to patroniPostgresCluster
	patroniPostgresCluster map[string]*marker
}

// marker is the tracker depends used to track endpoints created by patroni.
type marker struct {
	namespace string
	pgName    string
	//Cluster endpoint for postgres cluster cluster created by Patroni

	epsName string
	svcName string

	scheduledTime time.Time
}

// NewBackupTracker returns a BackupTracker
func NewPatroniTracker(deps *controller.Dependencies) PatroniTracker {
	tracker := &patroniTracker{
		deps:         deps,
		waitDuration: deps.CLIConfig.WaitDuration,
		//partoniEndpoints:       make(map[string]*trackDepends),
		patroniPostgresCluster: make(map[string]*marker),
	}
	return tracker
}

func (pt *patroniTracker) Length() int {
	return len(pt.patroniPostgresCluster)
}

func (pt *patroniTracker) StartTrackPostgresCluster(ns, pgName string) {
	pgkey := genPatroniResourcesKey(ns, pgName)
	if _, ok := pt.patroniPostgresCluster[pgkey]; !ok {
		scheduledTime := time.Now().Add(pt.waitDuration)

		mk := &marker{
			namespace: ns,
			pgName:    pgName,
			epsName:   pgName,
			svcName:   pgName + "-conifg",

			scheduledTime: scheduledTime,
		}

		pt.patroniPostgresCluster[pgkey] = mk
		klog.Infof("Register endpoints %s/%s and service %s/%s at scheduledTime %v", ns, mk.epsName, ns, mk.svcName, scheduledTime)
	}
}

func (pt *patroniTracker) StartTrackPatroniEndPoints() error {
	now := time.Now()
	for k, v := range pt.patroniPostgresCluster {
		ns := v.namespace
		pgName := v.pgName

		if now.Compare(v.scheduledTime) == -1 {
			klog.Infof("Check now %s and scheduledTime %v ", now, v.scheduledTime)
			continue
		}
		klog.Infof("Clean track resources at %v with scheduledTime %v ", time.Now(), v.scheduledTime)
		if !pt.isPatroniPostgresPodsExists(ns, pgName) {
			klog.Infof("Start to delete endpoints %s/%s created by Patroni ", ns, v.epsName)
			if v.epsName != "" {
				err := pt.deleteEndpoints(ns, v.epsName)
				if err != nil {
					klog.Infof("Failed to delete endpoints %s/%s, error: %v", ns, pgName, err)
					return err
				}
			}
			v.epsName = ""

			klog.Infof("Start to delete Service %s/%s created by Patroni ", ns, v.svcName)
			if v.svcName != "" {
				err := pt.deleteService(ns, v.svcName)
				if err != nil {
					klog.Infof("Failed to delete endpoints %s/%s, error: %v", ns, pgName, err)
					return err
				}
			}
			v.svcName = ""
			// remove marker for patroni PostgresCluster
			pt.removePartoniPostgresClusterMarker(k)
		}
	}
	klog.Infof("Patroni track return...")
	return nil

}

func (pt *patroniTracker) isPatroniPostgresPodsExists(namespace, pgName string) bool {

	podLabels := label.New().Instance(pgName).Patroni()
	pods, err := pt.deps.KubeClientset.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{
		LabelSelector: podLabels.String(),
	})
	if err != nil {
		klog.Errorf("Failed to list pods for PostgresCluster %s/%s, err: %v", namespace, pgName, err)
		return false
	}
	if len(pods.Items) == 0 {
		return false
	}
	return true

}

// removePartoniPostgresClusterMarker removes marker from Partoni PostgresCluster.
func (pt *patroniTracker) removePartoniPostgresClusterMarker(key string) {
	pt.operateLock.Lock()
	defer pt.operateLock.Unlock()
	delete(pt.patroniPostgresCluster, key)
}

func isEndpointCreatedByPartroni(endpoint corev1.Endpoints) bool {
	epLabels := endpoint.GetLabels()
	if _, ok := epLabels[label.PatroniApplicationLabelKey]; ok && len(epLabels) == 2 {
		if epLabels[label.PatroniClusterNameLabelKey] != "" {
			return true
		}
	}
	return false
}

// deleteEndpoints delete endpoints created by Patronio.
func (pt *patroniTracker) deleteEndpoints(ns, name string) error {
	err := pt.deps.KubeClientset.CoreV1().Endpoints(ns).Delete(context.TODO(), name, metav1.DeleteOptions{})
	if errors.IsNotFound(err) {
		klog.Infof("Endpoints %s/%s has been deleted, continue...", ns, name)
		return nil
	}
	return err
}

// deleteService delete Service created by Patronio.
func (pt *patroniTracker) deleteService(ns, name string) error {
	err := pt.deps.KubeClientset.CoreV1().Services(ns).Delete(context.TODO(), name, metav1.DeleteOptions{})
	if errors.IsNotFound(err) {
		klog.Infof("Service %s/%s has been deleted, continue...", ns, name)
		return nil
	}
	return err
}

func genPatroniResourcesKey(ns, name string) string {
	return fmt.Sprintf("%s/%s", ns, name)
}
